package com.xjscrm.console.chat.mq.consumer;

import com.alibaba.fastjson.JSON;
import com.uzai.trace.TraceGenerate;
import com.xjscrm.common.entity.WwCustomer;
import com.xjscrm.common.entity.WwGroupMember;
import com.xjscrm.console.chat.cache.CacheHolder;
import com.xjscrm.console.chat.enums.MQMsgType;
import com.xjscrm.console.chat.frame.dto.ChatRecordDTO;
import com.xjscrm.console.chat.frame.dto.ChatSessionDTO;
import com.xjscrm.console.chat.frame.send.SRecallSuccessMsg;
import com.xjscrm.console.chat.frame.send.SResultMsg;
import com.xjscrm.console.chat.mq.dto.DispatchDTO;
import com.xjscrm.console.chat.mq.dto.SendDTO;
import com.xjscrm.console.chat.mq.dto.SendMQDTO;
import com.xjscrm.console.chat.mq.dto.StaffOfflineDTO;
import com.xjscrm.console.chat.utils.ChannelUtils;
import com.xjscrm.console.chat.utils.MessageUtil;
import com.xjscrm.console.entity.ChatTalkRecord;
import com.xjscrm.console.entity.ChatTalkSession;
import com.xjscrm.console.entity.ChatTalkSessionStaffRel;
import com.xjscrm.console.mapper.WwGroupMemberMapper;
import com.xjscrm.console.service.customer.chat.ChatTalkRecordService;
import com.xjscrm.console.service.customer.chat.ChatTalkSessionService;
import com.xjscrm.console.service.customer.chat.ChatTalkSessionStaffRelService;
import com.xjscrm.console.service.customer.wwCustomer.WwCustomerService;
import io.netty.channel.Channel;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.MDC;
import org.springframework.beans.BeanUtils;
import org.springframework.messaging.support.MessageBuilder;

import java.util.List;
import java.util.Objects;

/**
 * @author mingyang.chen create in 2024/3/12 14:10
 */
@Slf4j
public class MsgSendConsumer implements MessageListenerConcurrently {

    private final ChatTalkSessionService chatTalkSessionService;
    private final ChatTalkRecordService chatTalkRecordService;
    private final CacheHolder cacheHolder;
    private final RocketMQTemplate rocketMQTemplate;
    private final WwCustomerService wwCustomerService;
    private final WwGroupMemberMapper wwGroupMemberMapper;
    private final ChatTalkSessionStaffRelService chatTalkSessionStaffRelService;

    public MsgSendConsumer(ChatTalkSessionService chatTalkSessionService, ChatTalkRecordService chatTalkRecordService, CacheHolder cacheHolder,
                           RocketMQTemplate rocketMQTemplate, WwCustomerService wwCustomerService, WwGroupMemberMapper wwGroupMemberMapper, ChatTalkSessionStaffRelService chatTalkSessionStaffRelService) {
        this.chatTalkSessionService = chatTalkSessionService;
        this.chatTalkRecordService = chatTalkRecordService;
        this.cacheHolder = cacheHolder;
        this.rocketMQTemplate = rocketMQTemplate;
        this.wwCustomerService = wwCustomerService;
        this.wwGroupMemberMapper = wwGroupMemberMapper;
        this.chatTalkSessionStaffRelService = chatTalkSessionStaffRelService;
    }

    @TraceGenerate
    public void onMessage(SendDTO message) {

        log.info("发送消息: 收到消息: {}", JSON.toJSONString(message));
        Long merId = message.getMerId();
        Long talkSessionId = message.getTalkSessionId();
        Long talkRecordId = message.getTalkRecordId();
        Long staffId = message.getStaffId();

        //
        ChatTalkSession talkSession = chatTalkSessionService.getById(merId, talkSessionId);
        ChatSessionDTO chatSessionDTO = new ChatSessionDTO();
        BeanUtils.copyProperties(talkSession, chatSessionDTO);
        //
        ChatTalkRecord talkRecord = chatTalkRecordService.getById(merId, talkRecordId);
        ChatRecordDTO chatRecordDTO = new ChatRecordDTO();
        if (Objects.nonNull(talkRecord)) {
            BeanUtils.copyProperties(talkRecord, chatRecordDTO);
            chatRecordDTO.parse();
        }
        ChatTalkSessionStaffRel sessionStaffRel = chatTalkSessionStaffRelService.getByTalkId(merId, talkSession.getDeviceUniqueId(), talkSessionId);
        if (Objects.nonNull(sessionStaffRel)) {
            chatSessionDTO.setHang(sessionStaffRel.getHang());
            chatSessionDTO.setTransfer(sessionStaffRel.getTransfer());
            chatSessionDTO.setAuto(sessionStaffRel.getAuto());
        }

        boolean isGroup = Objects.equals(talkSession.getType(), 111);

        WwCustomer customer;
        if (!isGroup) {
            customer = wwCustomerService.getByWxId(merId, talkSession.getDeviceUniqueId(), talkSession.getConversationId());
            if (Objects.nonNull(customer)) {
                talkSession.setConversationAvatar(customer.getHeadImgUrl());
            }
        } else {
            WwGroupMember member = wwGroupMemberMapper.getByGroupWxIdAndWxid(talkSession.getConversationId(), talkRecord.getSendWxId(), merId);
            if (Objects.nonNull(member)) {
                chatRecordDTO.setHeadImgUrl(member.getHimg());
                chatRecordDTO.setSendWxName(member.getNick());
            }
        }
        //
        chatTalkSessionService.fillTag(talkSession, tag -> {
            if (Objects.equals(tag.getType(), 1)) {
                chatSessionDTO.getCompanyTagIds().add(tag.getId());
            } else if (Objects.equals(tag.getType(), 1)) {
                chatSessionDTO.getSelfTagIds().add(tag.getId());
            }
        }, tag1 -> {
            chatSessionDTO.getSelfTagIds().add(tag1.getId());
        });

        Channel staffChannel = cacheHolder.getStaffChannel(staffId);
        if (Objects.isNull(staffChannel)) {
            cacheHolder.delStaffTag(staffId);
            cacheHolder.delStaffSessionCount(merId, staffId);
            cacheHolder.delStaffSessionNowCount(merId, staffId);

            DispatchDTO dispatchDTO = new DispatchDTO();
            dispatchDTO.setMerId(merId);
            dispatchDTO.setDeviceUniqueId(talkSession.getDeviceUniqueId());
            dispatchDTO.setTalkSessionId(talkSessionId);
            dispatchDTO.setTalkRecordId(talkRecordId);

            log.error("发送消息: 员工所在节点未上线, 重新分发: {}", JSON.toJSONString(dispatchDTO));
            //
            rocketMQTemplate.asyncSend("XJ_WW_DISPATCH_TOPIC", MessageBuilder.withPayload(dispatchDTO).build(), new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    log.info("发送消息: 重新分发消息发送成功 -> msg:{}", JSON.toJSONString(dispatchDTO));
                }

                @Override
                public void onException(Throwable throwable) {
                    log.error("发送消息: 重新分发消息发送失败", throwable);
                }
            });
            return;
        }
        cacheHolder.delSessionOrder(merId, talkSession.getDeviceUniqueId(), talkSessionId);
        cacheHolder.delSessionOrderRepeatTime(talkSession.getDeviceUniqueId(), talkSessionId);
        chatTalkSessionService.addChatSendDelayTask(merId, staffId, talkSession);
        MessageUtil.sendNewSessionTalk(staffChannel, chatSessionDTO, chatRecordDTO);
    }

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

        for (MessageExt msg : msgs) {
            String msgBody = new String(msg.getBody());
            SendMQDTO sendMQDTO = JSON.parseObject(msgBody, SendMQDTO.class);
            try {
                Integer type = sendMQDTO.getType();
                MQMsgType byCode = MQMsgType.getByCode(type);
                if (Objects.isNull(byCode)) {
                    continue;
                }
                switch (byCode) {
                    case push_talk:
                        SendDTO message = JSON.parseObject(sendMQDTO.getData().toString(), SendDTO.class);
                        MDC.put("traceId", message.getTraceId());
                        onMessage(message);
                        break;
                    case recall_success:
                        SRecallSuccessMsg sRecallSuccessMsg = JSON.parseObject(sendMQDTO.getData().toString(), SRecallSuccessMsg.class);
                        this.doSendRecallMsg(sRecallSuccessMsg);
                        break;
                    case staff_offline:
                        StaffOfflineDTO staffOfflineDTO = JSON.parseObject(sendMQDTO.getData().toString(), StaffOfflineDTO.class);
                        this.doStaffOffline(staffOfflineDTO);
                        break;
                }
            } catch (Exception e) {
                log.error("处理路由消息失败", e);
            } finally {
                MDC.clear();
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

    private void doStaffOffline(StaffOfflineDTO message) {
        log.info("收到员工下线消息: {}", JSON.toJSONString(message));

        Long merId = message.getMerId();

        Long staffId = message.getStaffId();
        Channel staffChannel = cacheHolder.getStaffChannel(staffId);
        if (Objects.nonNull(staffChannel)) {
            staffChannel.write(new TextWebSocketFrame(JSON.toJSONString(SResultMsg.fail(null, "当前员工已在其他地方登录, 即将关闭链接"))));
            ChannelUtils.setCloseAttr(staffChannel, "员工已在其他节点上线, 当前节点关闭");
            staffChannel.close();

            cacheHolder.delStaffSessionCount(merId, staffId);
            cacheHolder.delStaffChannel(staffId);
        }
    }

    private void doSendRecallMsg(SRecallSuccessMsg sRecallSuccessMsg) {
        Long r = sRecallSuccessMsg.getStaffId();
        Channel staffChannel = cacheHolder.getStaffChannel(r);
        if (Objects.nonNull(staffChannel)) {
            MessageUtil.sendMsg(staffChannel, sRecallSuccessMsg);
        }
    }
}
